package com.atguigu.userprofile.app;

import com.atguigu.userprofile.bean.TagInfo;
import com.atguigu.userprofile.constants.ConstCodes;
import com.atguigu.userprofile.dao.TagInfoDAO;
import com.atguigu.userprofile.utils.MyClickhouseUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.sql.SparkSession;

import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

public class TaskBitmapApp {
    public static void main(String[] args) {
        SparkConf sparkConf = new SparkConf().setAppName("TaskBitmapApp");
        SparkSession sparkSession = SparkSession.builder().config(sparkConf).enableHiveSupport().getOrCreate();

        //获取外部传参
        String taskId = args[0];
        String busiDate = args[1];

        //获取已启用的标签
        List<TagInfo> tagInfoWithOn = TagInfoDAO.getTagInfoWithOn();

        //创建四个存放不同类型的list集合
        ArrayList<TagInfo> tagInfoWithLongList = new ArrayList<>();
        ArrayList<TagInfo> tagInfoWithDecimalList = new ArrayList<>();
        ArrayList<TagInfo> tagInfoWithStringList = new ArrayList<>();
        ArrayList<TagInfo> tagInfoWithDateList = new ArrayList<>();

        //通过tagValueType进行判断，将相同标签值类型的tagInfo数据放一块
        for (TagInfo tagInfo : tagInfoWithOn) {
            if (ConstCodes.TAG_VALUE_TYPE_LONG.equals(tagInfo.getTagValueType())){
                tagInfoWithLongList.add(tagInfo);
            }else if (ConstCodes.TAG_VALUE_TYPE_DECIMAL.equals(tagInfo.getTagValueType())){
                tagInfoWithDecimalList.add(tagInfo);
            }else if (ConstCodes.TAG_VALUE_TYPE_STRING.equals(tagInfo.getTagValueType())){
                tagInfoWithStringList.add(tagInfo);
            }else if (ConstCodes.TAG_VALUE_TYPE_DATE.equals(tagInfo.getTagValueType())){
                tagInfoWithDateList.add(tagInfo);
            }
        }

        //调用方法执行sql
        executeSQL("user_tag_value_string", tagInfoWithStringList, busiDate);
        executeSQL("user_tag_value_long",tagInfoWithLongList,busiDate);
        executeSQL("user_tag_value_decimal",tagInfoWithDecimalList,busiDate);
        executeSQL("user_tag_value_date",tagInfoWithDateList,busiDate);

    }

    //TODO 注意！！！ 1.要判断集合中是否为空，不判断的话可能会在获取Tuple2元组元素的时候报错 2.关于uid的类型，在建表语句中us（存放uid的）中元素的类型要求是UInt64，那么就必须把uid转为UInt64 3.注意如果当天的数据重复跑的话，脏数据问题，可以在跑之前把当前的老数据先删掉


    //插入语句的方法
    public static void executeSQL(String tableName,List<TagInfo> tagInfoList,String busiDate){
        /**
         * insert into user_tag_value_string
         * select tp.1,tp.2,groupBitmapState(uid) from
         * (select uid,arrayJoin([('gender',gender),('agegroup',agegroup),('favor',favor)]) as tp from user_tag_merge) group by tp.1,tp.2;
         */
        if (tagInfoList.size()>0) {
            List<String> tagCodeWithFiledList = tagInfoList.stream().map(tagInfo -> "('" + tagInfo.getTagCode().toLowerCase() + "'," + tagInfo.getTagCode().toLowerCase() + ")").collect(Collectors.toList());

            String tagCodeWithFiledSQL = StringUtils.join(tagCodeWithFiledList, ",");
            System.out.println(tagCodeWithFiledSQL);

            //获取数据来源的表名
            String sourceTableName = "up_tag_merge_" + busiDate.replace("-", "");

            //编写查询语句
            String querySQL = "select tp.1,tp.2,groupBitmapState(cast(uid as UInt64)),'" + busiDate + "' from\n" +
                    " (select uid,arrayJoin([" + tagCodeWithFiledSQL + "]) as tp from " + sourceTableName + ") group by tp.1,tp.2";

            //幂等性处理 一旦当天跑的数据有问题的话，可以提前把之前老的数据先删掉 防止脏数据
            String deleteSQL = " alter table " + tableName + " delete where dt='" + busiDate + "'";

            //编写插入语句
            String insertSQL = " insert into " + tableName + " " + querySQL;

            //执行语句
            MyClickhouseUtil.executeSql(deleteSQL);
            MyClickhouseUtil.executeSql(insertSQL);
        }
    }
}
